// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#pragma once

#include <map>
#include <memory>
#include <set>
#include <string>
#include <vector>

#include "cc/coin2/exchange/di/constant.h"
#include "coin/feed/fastfeed/feed.pb.h"
#include "coin2/base/string_util.h"
#include "coin2/exchange/base/feed/system.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/feed.pb.h"
#include "coin2/feed/fastfeed/system.h"
#include "coin2/feed/navi/reader.h"

using coin2::exchange::base::market::ExchangeTypeToString;
using coin2::exchange::base::market::MarketExchangeApiRecipe;
using coin2::exchange::feed::FeedSystemConfig;
using coin2::feed::fastfeed::FastFeedUpdate;

class NaviFeedArchiveSystem : public coin2::exchange::base::feed::IFeedSubsystem,
                              public coin2::exchange::base::feed::FeedArchiveMixin {
 public:
  NaviFeedArchiveSystem(
      const std::string& system_name,
      // we use recipe as tag.
      // For KRX, recipe = ""
      // For china, recipe = "tag-barber_china_stock_csv_l1--book_depth-5"
      const MarketExchangeApi& mea,
      const coin2::exchange::di::ConstantInjection& ci,
      const FeedSystemConfig& config,
      coin2::exchange::base::executor::ArchiveSyncExecutor* ass,
      coin2::exchange::base::feed::IFeedSubscriber* callback)
      : IFeedSubsystem(system_name, mea), FeedArchiveMixin(ass), mea_(mea), config_(config) {
    AddCallback(callback);

    CHECK_EQ_THROW(config.type(), coin2::exchange::feed::FEED_SYSTEM_TYPE_NAVI_ARCHIVE);

    CHECK_EQ_THROW(
        config.archive().archive_type(),
        coin2::exchange::feed::ArchiveConfig_ArchiveType_FASTFEED);
    coin2::base::config::ValidateArchiveConfig(config.archive());

    const auto& archive = config_.archive();
    auto interval = coin2::base::config::FromIntervalConfig(archive.interval());
    auto warmup = DurationFromString(archive.read_warmup_duration());

    LOG(INFO) << "[NaviFeed Archive System]";
    LOG(INFO) << "              mea: " << mea_.String();
    LOG(INFO) << "interval - warmup: " << DebugStringFromDuration(warmup);
    LOG(INFO) << "interval -  start: " << Iso8601FromTimestamp(interval.first);
    LOG(INFO) << "interval -    end: " << Iso8601FromTimestamp(interval.second);

    for (const auto& s : config_.products().norms()) {
      symbols_.insert(s);
    }
    for (const auto& s : config_.products().natives()) {
      symbols_.insert(s);
    }

    // NaviFeedProtoReader is capable of handling multiple exchanges,
    // but here we use it for one exchange

    start_time_ = interval.first;

    if (archive.navi_feed_type() ==
        coin2::exchange::feed::ArchiveConfig_NaviFeedType::
            ArchiveConfig_NaviFeedType_NAVI_FEED_PER_SYMBOL) {
      reader_ = std::make_unique<NaviPerSymbolFeedReader>(
          archive.navi_per_symbol_root_dir(), archive.machine(),
          StringToUpperCopy(ExchangeTypeToString(mea.exchange)), symbols_,
          warmup, interval.first, interval.second, archive.max_record_size());
    } else if (archive.navi_feed_type() ==
               coin2::exchange::feed::ArchiveConfig_NaviFeedType::
                   ArchiveConfig_NaviFeedType_NAVI_FEED_UNIVERSE) {
      std::set<std::string> channels;
      for (const auto& c : config.channels()) {
        channels.insert(c);
      }
      reader_ = std::make_unique<NaviUniverseFeedReader>(
          archive.navi_universe_root_dir(), archive.machine(),
          archive.universe(), channels, warmup, interval.first, interval.second,
          archive.max_record_size());
    }

    update_ptr_.reset(new FastFeedUpdate(system_name, mea_));
  }

  //     feed_type: Book,
  //     data_type: BookSnapshot,
  //     symbol: "ADA-ETH.Huobi.adaeth",

  void Prepare() override {}
  void Cleanup() override {}

  int64_t PeekTimestamp() override {
    if (!has_feed_) {
      while (reader_->Read(&feed_)) {
        if (symbols_.size() > 0 &&
            symbols_.find(feed_.symbol()) == symbols_.end()) {
          continue;
        }

        feed_.set_timestamp(feed_.timestamp() + config_.archive().offset_ns());
        has_feed_ = true;
        break;
      }
      if (!has_feed_) {
        return 0;
      }
    }
    return feed_.timestamp();
  }

  void ConsumeRecord() override {
    using namespace ::fastfeed::proto;
    CHECK_THROW(has_feed_) << "Cannot consume a feed";
    has_feed_ = false;

    std::string symbol = feed_.symbol();
    FastFeedUpdate& update_ = *update_ptr_.get();
    update_.SetTimestamp(feed_.timestamp());
    // update_.SetProduct(); // please support non-coin KRX product
    update_.SetNativeSymbol(feed_.symbol());

    if (products_.find(symbol) == products_.end()) {
      products_[symbol].reset(new coin2::exchange::base::symbology::PseudoProduct(symbol));
    }
    update_.SetProduct(*products_[symbol].get());

    if (feed_.feed_type() == FEED_TYPE_BOOK) {
      BookBuilder* b = books_[symbol].get();
      if (b == nullptr) {
        BookBuilder::BookOption book_type{BookBuilder::SNAPSHOT};
        if (config_.raw_parser().enable_diff_book()) {
          book_type = BookBuilder::DELTA;
        } else if (config_.raw_parser().publish_bbo_book_only()) {
          book_type = BookBuilder::LIGHT_BBO;
        }
        books_[symbol].reset(b = new BookBuilder(book_type));
      }

      if (feed_.data_type() == DATA_TYPE_BOOK_SNAPSHOT) {
        const auto& snapshot = feed_.book_snapshot();
        auto tx = b->NewTransaction(feed_.origin_timestamp());
        tx.Clear();
        for (const auto& level : snapshot.ask()) {
          tx.UpdateAsk(level.price() / 10000.0f, level.qty());
        }
        for (const auto& level : snapshot.bid()) {
          tx.UpdateBid(level.price() / 10000.0f, level.qty());
        }
      } else if (feed_.data_type() == DATA_TYPE_BOOK_UPDATE) {
        const auto& diff = feed_.book_update();
        auto tx = b->NewTransaction(feed_.origin_timestamp());
        for (const auto& level : diff.ask()) {
          tx.UpdateAsk(level.price() / 10000.0f, level.qty());
        }
        for (const auto& level : diff.bid()) {
          tx.UpdateBid(level.price() / 10000.0f, level.qty());
        }
      }

      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      update_.SetFeedType(coin2::exchange::feed::BOOK_FEED);
      update_.SetBook(b);
      onBookFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_TRADE) {
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& trade = feed_.trade();
      update_.SetFeedType(coin2::exchange::feed::TRADE_FEED);
      update_.SetTrade(&trade_);
      trade_ = {
          feed_.origin_timestamp(),
          static_cast<double>(trade.price() / 10000.0f),
          static_cast<double>(trade.qty()),
          trade.side() == ::fastfeed::proto::TRADE_SIDE_BUY
              ? coin2::exchange::feed::TRADE_BUY_SIDE
              : coin2::exchange::feed::TRADE_SELL_SIDE,
          feed_.feed_continue() == ::fastfeed::proto::FEED_CONTINUE_YES,
      };
      onTradeFeed(update_);
    } else if (feed_.feed_type() == FEED_TYPE_LIQUIDATION_ORDER) {
      // skip warmup
      if (update_.timestamp() < start_time_) {
        return;
      }

      const auto& liquidation = feed_.liquidation();
      update_.SetFeedType(coin2::exchange::feed::STATUS_FEED);
      update_.SetLiquidationOrder(&liquidation_order_);
      coin::proto::OrderSide order_side = coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
      if (liquidation.direction() == ::fastfeed::proto::ORDER_DIRECTION_SELL) {
        order_side = coin::proto::OrderSide::SELL_ORDER;
      } else if (liquidation.direction() == ::fastfeed::proto::ORDER_DIRECTION_BUY) {
        order_side = coin::proto::OrderSide::BUY_ORDER;
      }
      liquidation_order_ = {
          feed_.origin_timestamp(),
          static_cast<double>(liquidation.price() / 10000.0f),
          static_cast<double>(liquidation.qty()),
          0,
          0,
          coin::proto::OrderType::UNKNOWN_ORDER_TYPE,
          order_side,
          coin::proto::OrderDuration::GTC_ORDER,
          coin::proto::OrderState::CREATED_ORDER,
          feed_.feed_continue() == ::fastfeed::proto::FEED_CONTINUE_YES,
          liquidation.order_id(),
      };
      onLiquidationFeed(update_);
    }
  }

  std::vector<coin2::exchange::base::symbology::IProduct*> GetProducts() const override {
    return {};
  }

 private:
  const MarketExchangeApi mea_;
  const FeedSystemConfig& config_;

  // states
  int64_t start_time_;
  std::unique_ptr<orio::io::Reader<CoinFeed>> reader_;
  bool has_feed_ = false;
  std::map<std::string, std::unique_ptr<BookBuilder>> books_;
  Trade trade_{};
  fastfeed::proto::Feed feed_;
  LiquidationOrder liquidation_order_{};
  std::unique_ptr<FastFeedUpdate> update_ptr_;

  std::set<std::string> symbols_;
  std::map<std::string, std::unique_ptr<coin2::exchange::base::symbology::IProduct>> products_;
};
